package com.spark.mooc.ch5_rdd.part04_cases

import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * @description: 有多个输入文件，每个文件中的每一行内容均为一个整数。
 *              要求读取所有文件中的整数，进行排序后，输出到一个新的文件中，输出的内容个数为每行两个整数，
 *              第一个整数为第二个整数的排序位次，第二个整数为原待排序的整数
 * @time: 2020/11/26 23:49
 * @author: lhy
 */
object FileSort {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("FileSort").setMaster("local")
        val sc = new SparkContext(conf)
        sc.setLogLevel("ERROR")     //设置日志格式级别
        val lines: RDD[String] = sc.textFile("input/rdd/fileSort/*", 3)
        var index = 0
        val result: RDD[(Int, Int)] = lines.filter(_.trim.length > 0).map(n => (n.trim.toInt,""))
                          .partitionBy(
                              new HashPartitioner(1)
                          ).sortByKey()
                           .map(t => {
                               index += 1
                               (index,t._1)
                           })
        result.foreach(println)
        result.saveAsTextFile("output/rdd/fileSort")
    }
}
